package com.atguigu.func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.common.GmallConfig;
import com.atguigu.uitl.DruidDSUtil;
import com.atguigu.uitl.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * @author hjy
 * @create 2023/3/9 16:04
 */
public class DimTableProcessFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject> {
    private MapStateDescriptor<String, TableProcess> stateDescriptor;

    public DimTableProcessFunction(MapStateDescriptor<String, TableProcess> stateDescriptor) {
        this.stateDescriptor = stateDescriptor;
    }
    // 这里的两个方法没有先后顺序，所以可能存在一种情况，当主流的数据来的时候，配置信息还在加载，这样就可能会将需要的数据过滤出去，所以要将配置信息读出一份来放到hashMap中,做一份保险
    //todo 在open方法中读取mysql中的配置信息放入hashMap
    private HashMap<String, TableProcess> hashMap;

    @Override
    public void open(Configuration parameters) throws Exception {
        hashMap = new HashMap<>();
        //从mysql中读取数据 需要用到JDBC连接，这里创建一个新的类，来适配所有的JDBC连接
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall_config?" +
                "user=root&password=123456&useUnicode=true&" +
                "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        //从mysql中获取数据
        List<TableProcess> tableProcessList = JdbcUtil.getJDBC(conn, "select * from table_process where sink_type='dim'", TableProcess.class, true);
        for (TableProcess tableProcess : tableProcessList) {
            hashMap.put(tableProcess.getSourceTable(),tableProcess);
        }
        conn.close();
    }

    //{"before":null,"after":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1669162876406,"snapshot":"false","db":"gmall-220623-config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1669162876406,"transaction":null}
    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        //获取状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(stateDescriptor);
        //todo 向状态写入数据
        // 拿到value的数据
        JSONObject jsonObject = JSONObject.parseObject(value);
        if ("d".equals(jsonObject.getString("op"))){
            TableProcess tableProcess = JSONObject.parseObject(jsonObject.getString("before"), TableProcess.class);
            if ("dim".equals(tableProcess.getSinkType())){
                String table = tableProcess.getSourceTable();
                broadcastState.remove(table);
                hashMap.remove(table);
            }
        }else {
            TableProcess tableProcess = JSONObject.parseObject(jsonObject.getString("after"), TableProcess.class);
            //建表
            createTable(tableProcess);
            if ("dim".equals(tableProcess.getSinkType())){
                String table = tableProcess.getSourceTable();
                ////写入状态
                broadcastState.put(table,tableProcess);
            }
        }
    }
    //create table if not exists xxx (id varchar primary key , name varchar) xxx
    private void createTable(TableProcess tableProcess)  {
        //拼接建表语句
        StringBuilder sql = new StringBuilder();
        sql.append("create table if not exists ")
                .append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(tableProcess.getSinkTable())
                .append(" ( ");
        String sinkColumns = tableProcess.getSinkColumns();
        String sinkPk = tableProcess.getSinkPk();
        String sinkExtend = tableProcess.getSinkExtend();
        if (sinkPk==null||"".equals(sinkPk)){
            sinkPk="id";
        }
        if (sinkExtend==null){
            sinkExtend="";
        }
        String[] fields = sinkColumns.split(",");
        int Columns = fields.length;
        for (int i = 0; i < Columns; i++) {
            sql.append(fields[i])
                    .append(" varchar ");
            if (fields[i].equals(sinkPk)){
                sql.append(" primary key ");
            }
            if (i==Columns-1){
                sql.append(")")
                        .append(sinkExtend);
            }else {
                sql.append(",");
            }
        }

        String createSQL = sql.toString();
        //打印建表语句
        System.out.println("建表语句："+createSQL);

        try {
            //连接phoenix
            DruidPooledConnection phoenixConn = DruidDSUtil.getPhoenixConn();

            //预编译
            PreparedStatement preparedStatement = phoenixConn.prepareStatement(createSQL);

            //执行
            preparedStatement.execute();

            phoenixConn.close();
        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("建表"+tableProcess.getSinkTable()+"失败！！！");
        }
    }

    //{"database":"gmall-220623-flink","table":"comment_info","type":"insert","ts":1669162958,"xid":1111,"xoffset":13941,"data":{"id":1595211185799847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,"spu_id":10,"order_id":987,"appraise":"1204","comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38","operate_time":null}}
    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //获取广播数据
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(stateDescriptor);
        //过滤  行列过滤
        //行过滤   要求表名在状态中存在  且type必须为三种的其中一种
        String table = value.getString("table");
        String type = value.getString("type");
        TableProcess tableProcess = broadcastState.get(table);
        TableProcess mapTableProcess = hashMap.get(table);
        //行过滤   要求表名在状态中存在  且type必须为三种的其中一种
        if ((tableProcess!=null||mapTableProcess!=null) &&("insert".equals(type)||"update".equals(type)||"bootstrap-insert".equals(type))){
            if (tableProcess==null){
                tableProcess=mapTableProcess;
            }
            //列过滤  要求这里的字段和状态中的字段要一致
            filterColumns(tableProcess,value.getJSONObject("data"));
            //写出数据
            value.put("sink_table",tableProcess.getSinkTable());
            out.collect(value);
        }else {
            System.out.println("该表"+table+"不存在！！或者操作类型为"+type);
        }

    }

    private void filterColumns(TableProcess tableProcess, JSONObject data) {
        //这里进行列过滤，把过滤掉的数据从data中删除
        String sinkColumns = tableProcess.getSinkColumns();
        String[] Columns = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(Columns);
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(next -> !columnList.contains(next.getKey()));
    }


}
